package com.joysuch.wwyt.mh.websocket;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.joysuch.wwyt.api.dataaccess.OrientationDataSyncService;
import com.joysuch.wwyt.api.dataaccess.bean.*;
import com.joysuch.wwyt.bp.entity.risk.BpCompany;
import com.joysuch.wwyt.core.common.util.SpringContextUtils;
import com.joysuch.wwyt.core.enums.RiskLevels;
import com.joysuch.wwyt.mh.bean.BpCompanyBean;
import com.joysuch.wwyt.mh.service.MonitorPointDataWebSocketService;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PreDestroy;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * 监测点实时数据&报警内容推送
 *
 * @Author ZhangDong
 * @Date 2021/2/3 9:43
 */
@ServerEndpoint(value = "/bp/risk/bigScreen/{tenentId}/{orgCode}/{subType}")
@Component
@Slf4j
@Data
public class MonitorPointDataWebSocket {
    private Session session;
    //生命终结点
    private volatile long activeTime;
    //最后一次ping 时间点, 4s内无应答累计次数
    private volatile long pingTime;
    //无应答次数，最多3次结束生命
    private AtomicInteger noRespCount = new AtomicInteger();
    private Map<Long, LocalDateTime> pingTimeMap = new ConcurrentHashMap<>();
    //所有连接 key: tenentId_orgCode_subType
    private static Map<String, CopyOnWriteArraySet<MonitorPointDataWebSocket>> webSocketMap = new ConcurrentHashMap<>(5);
    //ping允许的最大无响应时间,毫秒
    private static final int PING_NO_RESPONSE_MILLI_SECOND = 4000;
    //ping非正常回复累计次数
    private static final int PING_NO_RESPONSE_COUNT = 3;
    //pingTimeMap处理移除N秒钟之前的数据
    private static final int PING_TIME_HANDLE_REMOVE_SECONDS = 15;
    private static final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(5, r -> {
        Thread thread = new Thread(r, "monitor point worker");
        thread.setDaemon(true);
        return thread;
    });
    //公司风险研判缓存  key:tenentId_orgCode  value:bpCompany
    private static Map<String, BpCompany> bigScreenMap = new ConcurrentHashMap<>();

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(@PathParam(value = "tenentId") Long tenentId, @PathParam(value = "orgCode") String orgCode,
                       @PathParam(value = "subType") String subType, Session session) {
        if (!WebsocketSubType.existSubType(subType)) {
            return;
        }
        this.session = session;
        // session.setMaxTextMessageBufferSize(512000000);
        prolongLife();
        if (StringUtils.isNotEmpty(orgCode) && tenentId != null) {
            orgCode = orgCode.substring(0, 7);
            String key = getKey(tenentId, orgCode, subType);
            String bigScreenMapKey = tenentId + "_" + orgCode;
            synchronized (key.intern()) {
                CopyOnWriteArraySet<MonitorPointDataWebSocket> webSockets = webSocketMap.get(key);
                if (webSockets == null) {
                    webSockets = new CopyOnWriteArraySet<>();
                }
                webSockets.add(this);
                webSocketMap.put(key, webSockets);
            }
            session.getUserProperties().put("orgCode", orgCode);
            session.getUserProperties().put("tenentId", tenentId);
            session.getUserProperties().put("subType", subType);
            //sevice
            OrientationDataSyncService orientationDataSyncService = SpringContextUtils.getBean(OrientationDataSyncService.class);
            MonitorPointDataWebSocketService monitorPointDataWebSocketService = SpringContextUtils.getBean(MonitorPointDataWebSocketService.class);
            //连接成功 立即推送报警信息
            if (WebsocketSubType.MONITOR.getSubType().equals(subType)) {
                //危险源报警信息
                monitorPointDataWebSocketService.getPointAlarmMessage(null, "ALL");
            } else if (WebsocketSubType.RISK_COMPANY.getSubType().equals(subType)) {
                //一连接即推消息
                BpCompany bpCompany = bigScreenMap.get(bigScreenMapKey);
                BpCompanyBean bean = new BpCompanyBean();
                BeanUtil.copyProperties(bpCompany,bean);
                bean.setRiskLevelName(RiskLevels.findRiskLevelName(bpCompany.getRiskLevel()));
                sendData(key, JSON.toJSONString(bean));
            } else if (WebsocketSubType.RISK_DISTRICT.getSubType().equals(subType)) {
                //风险分区报警信息
                monitorPointDataWebSocketService.getDistrictAlarmMessage(null, "ALL");
            } else if (WebsocketSubType.CUSTOMIZE_MONITOR.getSubType().equals(subType)) {
                monitorPointDataWebSocketService.sendCustomizeMonitorPoint();
            } else if (WebsocketSubType.INVESTIGATION.getSubType().equals(subType)) {
                //巡检排查报警
                monitorPointDataWebSocketService.getInspectAlarmMessage(null, "ALL", tenentId, orgCode);
            } else if (WebsocketSubType.AI_ALARM.getSubType().equals(subType)) {
                //AI报警
                monitorPointDataWebSocketService.getAIAlarmMessage(null, "ALL", tenentId, orgCode);
            } else if (WebsocketSubType.WORKFLOW_VIEW.getSubType().equals(subType)) {
                // 作业票图标
                monitorPointDataWebSocketService.getWorkflowView(orgCode, tenentId, null);
            } else if (WebsocketSubType.LOCATE_SCREEN.getSubType().equals(subType)) {
                //人员列表
                List<OrientationScreenBinderBean> binderList = orientationDataSyncService.getScreenBinderData(tenentId, orgCode, "webSocket");
                if (!CollectionUtils.isEmpty(binderList)) {
                    MonitorPointDataWebSocket.sendInfo(binderList, WebsocketDataType.LOCATE_BINDER.getIndex());
                }
                //人员的区域列表
                List<OrientationScreenAreaPersonBean> areaPersonBeanList = orientationDataSyncService.getV3ScreenAreaPerson(tenentId, orgCode);
                if (!CollectionUtils.isEmpty(areaPersonBeanList)) {
                    MonitorPointDataWebSocket.sendInfo(areaPersonBeanList, WebsocketDataType.LOCATE_AREA_PERSON.getIndex());
                }
                //作业票报警分页
                List<OrientationPageAlarmCertificateOrAreaBean> certificateAlarmList = orientationDataSyncService.pageCertificateOrAreaAlarm(tenentId, orgCode, new OrientationSearchBean(0, 100, "pageCertificateAlarm"));
                if (!CollectionUtils.isEmpty(certificateAlarmList)) {
                    MonitorPointDataWebSocket.sendInfo(certificateAlarmList, WebsocketDataType.LOCATE_CERTIFICATE_ALARM.getIndex());
                }
                //区域人员报警分页
                List<OrientationPageAlarmCertificateOrAreaBean> areaAlarmList = orientationDataSyncService.pageCertificateOrAreaAlarm(tenentId, orgCode, new OrientationSearchBean(0, 100, "pageAreaAlarm"));
                if (!CollectionUtils.isEmpty(areaAlarmList)) {
                    MonitorPointDataWebSocket.sendInfo(areaAlarmList, WebsocketDataType.LOCATE_AREA_ALRAM.getIndex());
                }
                //车辆报警分页
                List<OrientationPageAlarmCertificateOrAreaBean> carAlarmList = orientationDataSyncService.pageCertificateOrAreaAlarm(tenentId, orgCode, new OrientationSearchBean(0, 100, "pagCarAlarm"));
                if (!CollectionUtils.isEmpty(carAlarmList)) {
                    MonitorPointDataWebSocket.sendInfo(carAlarmList, WebsocketDataType.LOCATE_CAR_ALARM.getIndex());
                }
                //报警总数
                OrientationAlarmCountBean alarmCountBean = orientationDataSyncService.alarmMessageCount(tenentId, orgCode);
                if (alarmCountBean != null) {
                    MonitorPointDataWebSocket.sendInfo(Lists.newArrayList(alarmCountBean), WebsocketDataType.LOCATE_ALRAM_COUNT.getIndex());
                }
            } else if (WebsocketSubType.RISK_DISTRICT_LIST.getSubType().equals(subType)) {
                monitorPointDataWebSocketService.getDistrictList(null, "ALL", orgCode, tenentId);
                monitorPointDataWebSocketService.getRiskPointCount(orgCode, tenentId);
            }
            log.info("一个监测点websocket连接进来了============");
        }
    }

    private static String getKey(Long tenentId, String orgCode, String subType) {
        return tenentId + "_" + orgCode + "_" + subType;
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(Session session) {
        String orgCode = (String) session.getUserProperties().get("orgCode");
        Long tenentId = (Long) session.getUserProperties().get("tenentId");
        String subType = (String) session.getUserProperties().get("subType");
        String key = getKey(tenentId, orgCode, subType);
        removeWebSocket(key);
        try {
            session.close();
        } catch (IOException e) {
            log.error("监测点websocket心跳检测，关闭异常：", e.getMessage());
        }
        log.info("监测点websocket离线。。。");
    }

    public void removeWebSocket(String key) {
        synchronized (key.intern()) {
            CopyOnWriteArraySet<MonitorPointDataWebSocket> webSockets = webSocketMap.get(key);
            webSockets.remove(this);
        }
    }

    /**
     * 收到客户端消息后调用的方法
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        if (message != null) {
            try {
                String sessionId = session.getId();
                MonitorWebsocketDto dto = JSON.parseObject(message, MonitorWebsocketDto.class);
                if (dto != null) {
                    long sendTime = dto.getSendTime();
                    String busType = dto.getBusType();
                    if (MonitorPointBusType.PING_PONG.getBusEnType().equals(busType)) {
                        prolongLife();
                        if (pingTime == 0) {
                            log.debug("监测点模块 pong,sessionId:{}, pingTime=0 return", sessionId);
                            return;
                        }
                        if (pingTimeMap.containsKey(sendTime)) {
                            log.debug("监测点模块 pong,sessionId:{}, 此pingTime:{} 已被处理，ignore===", sessionId, pingTime);
                            return;
                        }
                        if (pingTime == sendTime) {
                            long currentTimeMillis = System.currentTimeMillis();
                            if (currentTimeMillis - pingTime > PING_NO_RESPONSE_MILLI_SECOND) {
                                noRespCount.incrementAndGet();
                                pingTimeMap.put(sendTime, LocalDateTime.now());
                                pingTime = 0;
                                log.debug("监测点模块 pong,sessionId:{}, pingTime=sendTime:{},当前时间戳:{}, pingTime清零 超过4s应答，次数加1，当前次数:{}",
                                        sessionId, sendTime, currentTimeMillis, noRespCount.get());
                                return;
                            }
                            pingTime = 0;
                            noRespCount.set(0);
                            log.debug("监测点模块 pong,sessionId:{}, pingTime=sendTime:{},pingTime清零 没超过4s应答，次数清零", sessionId, sendTime);
                        } else {
                            noRespCount.incrementAndGet();
                            log.debug("监测点模块 pong,sessionId:{}, pingTime:{} !=sendTime:{},次数加1，当前次数:{}",
                                    sessionId, pingTime, sendTime, noRespCount.get());
                        }
                    }
                }
            } catch (Exception e) {
            }
        }
    }

    private void prolongLife() {
        //续命5分钟
        activeTime = System.currentTimeMillis() + 5 * 60 * 1000;
    }

    /**
     * 发生错误时调用
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("监测点websocket连接出错：" + error.getMessage(), error);
    }

    /**
     * 群发消息
     * flag: 0 监测点数据; 1 危险源报警; 2 风险研判大屏,3风险分区报警,4自定义监测点 5人员列表 6人员的区域列表  7作业票报警列表 8区域人员报警列表 9报警总数 10巡检排查列表
     */
    public static void sendInfo(List<?> list, int flag) {
        //报警全部处理完 需要返回空值清空大屏报警
        if (CollectionUtils.isEmpty(list)) {
            if (flag == WebsocketDataType.LOCATE_CERTIFICATE_ALARM.getIndex()) {
                List<OrientationPageAlarmCertificateOrAreaBean> certificateList = new ArrayList<>();
                MonitorWebsocketDto dto = new MonitorWebsocketDto(MonitorPointBusType.LOCATE_SCREEN_CERTIFICATE_ALARM.getBusEnType(), certificateList);
                sendData(getKey(1L, "001-001", WebsocketSubType.LOCATE_SCREEN.getSubType()), JSON.toJSONString(dto));
            } else if (flag == WebsocketDataType.LOCATE_AREA_ALRAM.getIndex()) {
                List<OrientationPageAlarmCertificateOrAreaBean> certificateList = new ArrayList<>();
                MonitorWebsocketDto dto = new MonitorWebsocketDto(MonitorPointBusType.LOCATE_SCREEN_AREA_ALARM.getBusEnType(), certificateList);
                sendData(getKey(1L, "001-001", WebsocketSubType.LOCATE_SCREEN.getSubType()), JSON.toJSONString(dto));
            } else if (flag == WebsocketDataType.LOCATE_ALRAM_COUNT.getIndex()) {
                List<OrientationAlarmCountBean> areaList = new ArrayList<>();
                OrientationAlarmCountBean orientationAlarmCountBean = new OrientationAlarmCountBean();
                orientationAlarmCountBean.setAllCount(0);
                areaList.add(orientationAlarmCountBean);
                MonitorWebsocketDto dto = new MonitorWebsocketDto(MonitorPointBusType.LOCATE_SCREEN_ALARM_COUNT.getBusEnType(), areaList);
                sendData(getKey(1L, "001-001", WebsocketSubType.LOCATE_SCREEN.getSubType()), JSON.toJSONString(dto));
            }else if (flag ==WebsocketDataType.LOCATE_CAR_ALARM.getIndex()){
                List<OrientationPageAlarmCertificateOrAreaBean> certificateList = new ArrayList<>();
                MonitorWebsocketDto dto = new MonitorWebsocketDto(MonitorPointBusType.LOCATE_SCREEN_CAR_ALARM.getBusEnType(), certificateList);
                sendData(getKey(1L, "001-001", WebsocketSubType.LOCATE_SCREEN.getSubType()), JSON.toJSONString(dto));
            }
            return;
        }
        WebSocketCovertDataFactory factory = new WebSocketCovertDataFactory(flag, list);
        factory.covertData(flag);
        log.debug("监测点实时数据推送end ================");
    }

    /**
     * 发送数据
     */
    public static void sendData(String key, String message) {
        CopyOnWriteArraySet<MonitorPointDataWebSocket> webSocketSet = webSocketMap.get(key);
        if (webSocketSet == null || CollectionUtils.isEmpty(webSocketSet)) {
            return;
        }
        executor.execute(() -> {
            try {
                for (MonitorPointDataWebSocket webSocket : webSocketSet) {
                    try {
                        webSocket.sendMessage(message);
                    } catch (IOException e) {
                        log.error("监测点实时数据推送报错:" + e.getMessage());
                    }
                }
            } catch (Exception e) {
                log.error("监测点实时数据推送报错:" + e.getMessage());
                e.printStackTrace();
            }
        });
    }


    public void sendMessage(String message) throws IOException {
        synchronized (session) {
            if (session != null && session.isOpen()) {
                session.getBasicRemote().sendText(message);
            }
        }
    }

    @Scheduled(initialDelay = 2450, fixedDelay = 5000)
    public void checkHeart() {
        Collection<CopyOnWriteArraySet<MonitorPointDataWebSocket>> set = webSocketMap.values();
        if (CollectionUtils.isEmpty(set)) {
            return;
        }
        for (CopyOnWriteArraySet<MonitorPointDataWebSocket> webSockets : set) {
            for (MonitorPointDataWebSocket webSocket : webSockets) {
                checkHeartDetail(webSocket);
            }
        }
    }

    private void checkHeartDetail(MonitorPointDataWebSocket webSocket) {
        executor.execute(() -> {
            try {
                Session session = webSocket.getSession();
                String sessionId = session.getId();
                long pingTime = webSocket.getPingTime();
                AtomicInteger noRespCount = webSocket.getNoRespCount();
                long activeTime = webSocket.getActiveTime();
                Map<Long, LocalDateTime> pingTimeMap = webSocket.getPingTimeMap();
                removeOldPingTime(pingTimeMap);
                //监测生命体征
                if (System.currentTimeMillis() > activeTime || noRespCount.get() >= PING_NO_RESPONSE_COUNT) {
                    log.debug("监测点模块 次数达到3次 关闭连接");
                    onClose(session);
                    return;
                }
                if (pingTime != 0 && !pingTimeMap.containsKey(pingTime)
                        && System.currentTimeMillis() - pingTime > PING_NO_RESPONSE_MILLI_SECOND) {
                    pingTimeMap.put(pingTime, LocalDateTime.now());
                    noRespCount.incrementAndGet();
                    log.debug("监测点模块 pingTime != 0 自动监测应答时间超过4s，次数加1 ，当前次数：{}", webSocket.getNoRespCount().get());
                }
                //定时监测心跳
                if (session != null && session.isOpen()) {
                    try {
                        long currentTimeMillis = System.currentTimeMillis();
                        MonitorWebsocketDto dto = new MonitorWebsocketDto(MonitorPointBusType.PING_PONG.getBusEnType(), currentTimeMillis, "ping");
                        session.getBasicRemote().sendText(JSON.toJSONString(dto));
                        webSocket.setPingTime(currentTimeMillis);
                        log.debug("监测点模块 ping,sessionId:{}, pingTime:{}", sessionId, currentTimeMillis);
                    } catch (IOException e) {
                        log.error("监测点实时数据推送--心跳监测异常:{}", e.getMessage());
                    }
                }
            } catch (Exception e) {
                log.error("监测点-- 心跳监测异常:", e.getMessage());
            }
        });
    }

    private void removeOldPingTime(Map<Long, LocalDateTime> pingTimeMap) {
        for (Long pingTime : pingTimeMap.keySet()) {
            LocalDateTime localDateTime = pingTimeMap.get(pingTime);
            if (LocalDateTime.now().isAfter(localDateTime.plusSeconds(PING_TIME_HANDLE_REMOVE_SECONDS))) {
                log.debug("监测点模块 移除过期pinTime:{}，当前时间:{}, put时间:{}", pingTime, LocalDateTime.now(), localDateTime);
                pingTimeMap.remove(pingTime);
            }
        }
    }

    public static Map getScreenMap() {
        return bigScreenMap;
    }

    public static void putBigScreenMap(BpCompany bc) {
        String orgCode = bc.getOrgCode().substring(0, 7);
        Long tenentId = bc.getTenentId();
        String key = tenentId + "_" + orgCode;
        if (orgCode != null) {
            bigScreenMap.put(key, bc);
        }
    }

    //10点更新，当天没有公告需要清除数据
    public static void resetWebsocketContent(List<BpCompany> list) {
        Set<BpCompany> allSet = bigScreenMap.values().stream().collect(Collectors.toSet());
        Set<BpCompany> collect = list.stream().collect(Collectors.toSet());
        Sets.SetView<BpCompany> difference = Sets.difference(allSet, collect);
        for (BpCompany bpCompany : difference) {
            String orgCode = bpCompany.getOrgCode();
            log.info("orgCode:" + orgCode + "，当天无大屏数据，执行清除操作");
            BpCompany temp = new BpCompany();
            temp.setTenentId(bpCompany.getTenentId());
            temp.setOrgCode(orgCode);
            putBigScreenMap(temp);
            sendInfo(Lists.newArrayList(temp), WebsocketDataType.RISK_COMPANY.getIndex());
        }
    }

    @PreDestroy
    public void shutdown() {
        Collection<CopyOnWriteArraySet<MonitorPointDataWebSocket>> set = webSocketMap.values();
        if (CollectionUtils.isEmpty(set)) {
            return;
        }
        for (CopyOnWriteArraySet<MonitorPointDataWebSocket> webSockets : set) {
            for (MonitorPointDataWebSocket webSocket : webSockets) {
                Session session = webSocket.getSession();
                if (session != null && session.isOpen()) {
                    try {
                        session.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
        executor.shutdownNow();
    }

}
